package com.eric.r2d2.scheduler;

import java.net.UnknownHostException;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.eric.r2d2.pageProcessor.SimHash;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.MongoException;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;

import us.codecraft.webmagic.Request;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.scheduler.DuplicateRemovedScheduler;
import us.codecraft.webmagic.scheduler.MonitorableScheduler;
import us.codecraft.webmagic.scheduler.component.DuplicateRemover;

public class MongoScheduler extends DuplicateRemovedScheduler implements MonitorableScheduler, DuplicateRemover {

	private MongoDatabase db_scheduler;
	private MongoCollection<BasicDBObject> coll_url_scheduler;
	private MongoClient mc;

	private static BlockingQueue<String> tmp_poll_queue = new LinkedBlockingQueue<String>();

	private Logger logger = LoggerFactory.getLogger(getClass());

	public MongoScheduler(String host, String db, String coll) throws UnknownHostException {
		try {
			this.mc = new MongoClient(new MongoClientURI(host));
		} catch (MongoException e) {
			e.printStackTrace();
		}
		this.db_scheduler = mc.getDatabase(db);
		this.coll_url_scheduler = this.db_scheduler.getCollection(coll, BasicDBObject.class);
	}

	public synchronized Request poll(Task task) {
		if (tmp_poll_queue.isEmpty()) {
			logger.info(this.getClass() + "Require url queue from scheduler");
			// 从mongodb读取url，存入临时队列
			MongoCursor<BasicDBObject> c = coll_url_scheduler.find(new BasicDBObject().append("crawled", "no"))
					.limit(50).iterator();
			while (c.hasNext()) {
				BasicDBObject entry = c.next();
				String url = entry.getString("url");
				tmp_poll_queue.add(url);

				// 将已取出的url，设为crawled:yes
				BasicDBObject updateEntry = (BasicDBObject) entry.clone();
				updateEntry.put("crawled", "yes");
				BasicDBObject search = new BasicDBObject();
				search.put("url_hash", new SimHash(url, 64).simHash().toString());

				coll_url_scheduler.replaceOne(search, updateEntry);
			}
		}
		// 从临时队列返回Request
		return new Request(tmp_poll_queue.poll());
	}

	protected void pushWhenNoDuplicate(Request request, Task task) {
		String url = request.getUrl();
		BasicDBObject d = new BasicDBObject();
		d.put("url", url);
		d.put("crawled", "no");
		d.put("crawl_date", new Date());
		d.put("url_hash", new SimHash(url, 64).simHash().toString());

		try {
			coll_url_scheduler.insertOne(d);
		} catch (com.mongodb.MongoWriteException e) {

		}

	}

	public boolean isDuplicate(Request request, Task task) {
		String url = request.getUrl();
		MongoCursor<BasicDBObject> c = coll_url_scheduler.find(new Document("url", url)).iterator();
		if (c.hasNext())
			return true;
		return false;
	}

	public void resetDuplicateCheck(Task task) {
		// TODO Auto-generated method stub

	}

	public int getLeftRequestsCount(Task task) {
		// TODO Auto-generated method stub
		return 0;
	}

	public int getTotalRequestsCount(Task task) {
		// TODO Auto-generated method stub
		return 0;
	}

}
